package com.hz.ruoyimqttconsume.config;


import com.hz.ruoyimqttconsume.handler.MQTTMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * @author 侯文波
 * @date 2025/3/10 17:34
 * @description: mqtt 消费者配置
 */
@Configuration
@Slf4j
public class MQTTConsumeConfig {
    @Resource
    private MQTTConfig mqttConfig;

    @Resource
    private MQTTMessageHandler mqttMessageHandler;

    // 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
    private static final byte[] WILL_DATA;

    static {
        WILL_DATA = "offline".getBytes();
    }

    /**
     * @param
     * @Description 入站直连通道
     * @Throws
     * @Return org.springframework.messaging.MessageChannel
     * @Author 侯文波
     */
    @Bean("mqttInput")
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }


    /**
     * @param
     * @Description 创建MqttPahoClientFactory 设置MQTT的broker的连接属性
     * @Throws
     * @Return org.springframework.integration.mqtt.core.MqttPahoClientFactory
     * @Author 侯文波
     */
    @Bean
    public MqttPahoClientFactory inClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttConfig.getHost());
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setConnectionTimeout(mqttConfig.getTimeout());
        options.setKeepAliveInterval(mqttConfig.getKeepalive());
        options.setCleanSession(mqttConfig.getCleanSession());
        options.setAutomaticReconnect(mqttConfig.getAutomaticReconnect());

        // 设置“遗嘱”消息 设置“遗嘱”消息的话题，若客户端与Broker断开连接，此时该客户端publish一个消息到该主题，Broker会自动发布该消息至订阅该主题的客户端。
        options.setWill("willTopic", WILL_DATA, 2, false);
        factory.setConnectionOptions(options);
        return factory;
    }


    /**
     * @param
     * @Description 入站  创建消费者实例
     * @Throws
     * @Return org.springframework.integration.core.MessageProducer
     * @Author 侯文波
     */
    @Bean
    public MessageProducer producer() throws UnknownHostException {

        InetAddress localHost = InetAddress.getLocalHost();
        String ipAddress = localHost.getHostAddress();

        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + ipAddress.replace(".", "_") + "_customer", inClientFactory(), mqttConfig.getTopics());
        adapter.setCompletionTimeout(5000);
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        // 按字节接收消息
        // defaultPahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(defaultPahoMessageConverter);
        adapter.setQos(mqttConfig.getQos());
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }

    /**
     * @param
     * @Description 通过通道获取数据
     * ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
     * tips:
     * 异步处理
     * @Throws
     * @Return org.springframework.messaging.MessageHandler
     * @Author 侯文波
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInput")
    public MessageHandler handler() {
        return this.mqttMessageHandler;
    }
}
